package my.thesis.bolts.filtersReliable;

import java.util.Map;

import my.thesis.analytics.metrics.ThroughputMetric;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class FilterLanguageReliable extends BaseRichBolt
{
	private static final long serialVersionUID = 1L;
	private OutputCollector collector;
	
	transient ThroughputMetric _throughputMetric;
	private static Boolean metricsOn=false;
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) 
	{
		this.collector=collector;
		
		if(metricsOn)
		{
			_throughputMetric=new ThroughputMetric();
			context.registerMetric("filterLanguageCount", _throughputMetric, 20);
			_throughputMetric.setStartTime();
		}
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		declarer.declare(new Fields("id","country","lat","lon","text"));  		
	}

	public void execute(Tuple tuple) 
	{
		if(metricsOn)
		{
			_throughputMetric.incrTuples();
		}
		
		Long id = tuple.getLongByField("id");
		Double lat = tuple.getDoubleByField("lat");
		Double lon = tuple.getDoubleByField("lon");
		
		String country = tuple.getStringByField("country");
				
		String text=tuple.getStringByField("text");
		
		String lang=tuple.getStringByField("lang");

		if(lang.equals("en") && lat!=0 && lon!=0)
		{
			collector.emit(tuple,new Values(id,country,lat,lon,text));
		}
		
		collector.ack(tuple);
	}

}
